package com.ruoyi.common.utils.rabbitmq;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * @author xudengyu
 */
@Component
@Slf4j
public class RabbitMQUtil {

    private static final long DEFAULT_TIMEOUT = 1000L;

    public static boolean sendMessage(RabbitTemplate rabbitTemplate,
                                      String exchange,
                                      String routingKey,
                                      Object message,
                                      CorrelationData correlationData) throws InterruptedException {
        return sendMessage(rabbitTemplate, exchange, RabbitMQDynamicConfig.ROUTING_KEY_PREFIX+routingKey, message, correlationData, DEFAULT_TIMEOUT);
    }

    public static boolean sendMessage(RabbitTemplate rabbitTemplate,
                                      String exchange,
                                      String routingKey,
                                      Object message,
                                      CorrelationData correlationData,
                                      long timeoutMillis) throws InterruptedException {
        final CountDownLatch latch = new CountDownLatch(1);
        final boolean[] result = {false};

        correlationData.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {
            @Override
            public void onFailure(Throwable ex) {
                handleFailure(ex);
                result[0] = false;
                latch.countDown();
            }

            @Override
            public void onSuccess(CorrelationData.Confirm confirm) {
                result[0] = handleConfirm(confirm);
                latch.countDown();
            }
        });

        rabbitTemplate.convertAndSend(exchange, routingKey, message, correlationData);
        if (!latch.await(timeoutMillis, TimeUnit.MILLISECONDS)) {
            handleTimeout(correlationData.getId());
        }
        return result[0];
    }

    private static void handleFailure(Throwable ex) {
        log.error("### [消息发送] 业务异常：{}", ex.getMessage());
    }

    private static boolean handleConfirm(CorrelationData.Confirm confirm) {
        if (confirm.isAck()) {
            log.info("### [消息发送] 成功，收到ack");
            return true;
        } else {
            log.error("### [消息发送] 失败，收到nack。原因：{}", confirm.getReason());
            return false;
        }
    }

    private static void handleTimeout(String correlationId) {
        log.error("### [消息发送] 超时，correlationId: {}", correlationId);
    }




}
